package com.htdata.iiot.kafka2tsdb.kafka.muil;

import com.htdata.iiot.kafka2tsdb.cmd.Cmd;
import com.htdata.iiot.kafka2tsdb.config.Configs;
import org.apache.log4j.Logger;
import java.util.concurrent.atomic.LongAdder;

public class DstatusStore {
	private static Logger logger = Logger.getLogger(DstatusStore.class);
	public static LongAdder longAdder = new LongAdder();
	public static LongAdder metricAdder = new LongAdder();
	public static boolean log=false;
	public static void main(String[] args) {
		
		Cmd.parserCmd(args);
		
		long startTime = System.currentTimeMillis();
		logger.info("数据解析程序已启动。。。");
		for(int i=0;i<Configs.BaseConfig.threads;i++){
			new Thread(new MuilKafkaClient(i)).start();
		}
		
		new Thread(new Runnable() {
			private long oldcount=0;
			@Override
			public void run() {
				while(true){
					try {
						Thread.sleep(10000);
					} catch (InterruptedException e) {
						// TODO Auto-generated catch block
						e.printStackTrace();
					}
					long count = longAdder.sum();
					long mcount = metricAdder.sum();
					long endTime = System.currentTimeMillis();
					long thisTime = endTime-startTime;
					logger.info("获取到数据:"+count+"\t平均速率:"+(count*1000/thisTime)+"\t测点数:"+mcount+"\t测点速率:"+(mcount*1000/thisTime)+" \t this time speed :"+((count-oldcount)/10));
					oldcount=count;
				}
			}
		}).start();
	}

}
